package com.fafu.mq.mqserver.datacenter;

import com.fafu.mq.common.MqException;
import com.fafu.mq.mqserver.core.Binding;
import com.fafu.mq.mqserver.core.Exchange;
import com.fafu.mq.mqserver.core.MSGQueue;
import com.fafu.mq.mqserver.core.Message;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 使用的这个类来统一管理内存中的所有数据
 * 该类后续提供的一些方法, 可能会在多线程下被使用, 要注意线程安全问题
 */


public class MemoryDataCenter {
    // key 是 exchangeName, value 是 exchange 对象
    private ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();
    // key 是 queueName , value 是 MSCQueue 对象
    private ConcurrentHashMap<String, MSGQueue> queueMap = new ConcurrentHashMap<>();
    // 第一个 key 是 exchangeName , 第二个 key 是 queueName
    private ConcurrentHashMap<String,ConcurrentHashMap<String, Binding>> bindingsMap = new ConcurrentHashMap<>();
    // key 是 messageId , value 是 message 对象
    private ConcurrentHashMap<String, Message> messageMap = new ConcurrentHashMap<>();
    // key 是 queueName, value 是一个 message 消息的链表
    private ConcurrentHashMap<String, LinkedList<Message>> queueMessageMap = new ConcurrentHashMap<>();

    //第一个 key 是队列的名字, 第二个 key 是 messageId
    private ConcurrentHashMap<String,ConcurrentHashMap<String,Message>> queueMessageWaitAckMap = new ConcurrentHashMap<>();

    public void insertExchange(Exchange exchange){
        exchangeMap.put(exchange.getName(),exchange);
        System.out.println("[MemoryDataCenter] 新交换机添加成功!  exchangeName = " +
                exchange.getName());
    }

    public void deleteExchange(String exchangeName){
        exchangeMap.remove(exchangeName);
        System.out.println("[MemoryDataCenter] 新交换机删除成功!  exchangeName = " +
                exchangeName);
    }

    public Exchange getExchange(String exchangeName){
        return exchangeMap.get(exchangeName);
    }

    public void insertQueue(MSGQueue queue){
        queueMap.put(queue.getName(),queue);
        System.out.println("[MemoryDataCenter] 新队列添加成功!  queueName = " +
                queue.getName());
    }

    public void deleteQueue(String queueName){
        queueMap.remove(queueName);
        System.out.println("[MemoryDataCenter] 新队列删除成功!  queueName = " +
                queueName);
    }

    public MSGQueue getQueue(String queueName){
        return queueMap.get(queueName);
    }

    public void insertBinding(Binding bindIng) throws MqException {
        // 先判断是否存在这个哈希表, 不存在就创建一个
//        ConcurrentHashMap<String,BindIng> bindingMap = bindingsMap.get(bindIng.getExchangeName());
//        if (bindingMap == null){
//            bindingMap = new ConcurrentHashMap<>();
//            bindingMap.put(bindIng.getQueueName(),bindIng);
//        }
        // 以上代码有点不优雅, 我们换一种写法, 类似于 getOrDefault
        // 先根据 exchangeName 查一下, 看一下存不存在, 不存在就创建一个
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(bindIng.getExchangeName(),
                k -> new ConcurrentHashMap<>());
        synchronized (bindingMap){
            //再根据 queueName 查一下, 看一下存不存在 binding, 存在就直接抛出异常
            if(bindingMap.get(bindIng.getQueueName()) != null){
                throw new MqException("[MemoryDataCenter] 绑定已存在! exchangeName = " + bindIng.getExchangeName()
                        + ", queueName = " + bindIng.getQueueName());
            }
            bindingMap.put(bindIng.getQueueName(),bindIng);
//        bindingsMap.put(bindIng.getExchangeName(),bindingMap);
        }
        System.out.println("[MemoryDataCenter] 新Binding添加成功!  exchangeName = " +
                bindIng.getExchangeName() + " queueName = " + bindIng.getBindingKey());
    }

    //获取绑定, 我们写出两个版本
    // 1. 根据 exchangeName 和 queueName
    // 2. 根据 exchangeName 获取到了所有的 Binding
    public Binding getBinding(String exchangeName, String queueName){
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(exchangeName);
        if (bindingMap == null){
            return null;
        }
        return bindingMap.get(queueName);
    }

    public ConcurrentHashMap<String, Binding> getBindings(String exchangeName){
        return bindingsMap.get(exchangeName);
    }

    // 删除 binding
    public void deleteBinding(Binding bindIng) throws MqException {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(bindIng.getExchangeName());
        if (bindingMap == null) {
            throw new MqException("[MemoryDataCenter] 绑定不存在! exchangeName = " + bindIng.getExchangeName()
                    + ", queueName = " + bindIng.getQueueName());
        }
        bindingMap.remove(bindIng.getQueueName());
        System.out.println("[MemoryDataCenter] Binding删除成功!  exchangeName = " +
                bindIng.getExchangeName() + " queueName = " + bindIng.getBindingKey());
    }

    // 添加消息
    public void addMessage(Message message){
        messageMap.put(message.getMessageId(),message);
        System.out.println("[MemoryDataCenter] 新消息添加成功!  messageId = " +
                message.getMessageId());
    }

    // 根据 id 查询消息
    public Message getMessageById(String messageId){
        return messageMap.get(messageId);
    }

    // 根据 id 删除消息
    public void deleteMessage(String messageId){
        messageMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 消息删除成功!  messageId = " +
                messageId);
    }

    // 发送消息到指定队列
    public void sendMessage(MSGQueue queue,Message message){
        // 把消息放到对应的队列数据结构中
        // 根据队列的名字, 找到该队列的消息链表
//        LinkedList<Message> messages = queueMessageMap.get(queue.getName());
//        if (messages == null){
//            messages = new LinkedList<>();
//            queueMessageMap.put(queue.getName(),messages);
//        }
        // 替代
        // computeIfAbsent 是线程安全的
        LinkedList<Message> messages = queueMessageMap.computeIfAbsent(queue.getName(),
                k -> new LinkedList<>());
        // 再把数据加到 messages 中
        synchronized (messages){
            messages.add(message);
        }
        // 在这里我们把该消息直接插入一下总的哈希表 messageMap
        // 后者覆盖前者如果存在, 但是前后者是一样的
        addMessage(message);
        System.out.println("[MemoryDataCenter] 消息被投递到队列中成功!  messageId = " +
                message.getMessageId());
    }

    //从队列中取消息
    public Message pollMessage(String queueName){
        // 根据队列名, 查找一下, 对应的队列的消息链表
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if (messages == null){
            return null;
        }
        synchronized (messages){
            if(messages.size() == 0){
                return null;
            }
            // 链表中有元素, 进行头删
            Message currentMessage = messages.remove(0);
            System.out.println("[MemoryDataCenter] 消息从队列中取出!  messageId = " +
                    currentMessage.getMessageId());
            return currentMessage;
        }
    }

    //获取指定队列中消息的个数
    public int getMessageCount(String queueName){
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if(messages == null){
            return 0;
        }
        synchronized (messages){
            return messages.size();
        }
    }

    // 添加未确认的消息
    public void addMessageWaitAck(String queueName,Message message){
        ConcurrentHashMap<String,Message> messageHashMap = queueMessageWaitAckMap.computeIfAbsent(queueName,
                k -> new ConcurrentHashMap<>());
        messageHashMap.put(message.getMessageId(),message);
        System.out.println("[MemoryDataCenter] 消息从进入确认队列!  messageId = " +
                message.getMessageId());
    }

    // 删除未确认的消息(消息已确认)
    public void deleteMessageWaitAck(String queueName,String messageId){
        ConcurrentHashMap<String,Message> messageHashMap = queueMessageWaitAckMap.get(queueName);
        if (messageHashMap == null){
            return;
        }

        messageHashMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 消息从待确认队列删除!  messageId = " +
                messageId);

    }

    // 获取指定的未确认消息
    public Message getMessageWaitAck(String queueName,String messageId){
        ConcurrentHashMap<String,Message> messageHashMap = queueMessageWaitAckMap.get(queueName);
        if (messageHashMap == null){
            return null;
        }
        return messageHashMap.get(messageId);
    }


    // 这个方法就是从内存中读取数据, 把硬盘中之前持久化存储的各个维度的数据都恢复到内存中
    public void recovery(DiskDataCenter diskDataCenter) throws IOException, MqException, ClassNotFoundException {
        // 1. 清空原来的数据
        exchangeMap.clear();
        queueMap.clear();
        bindingsMap.clear();
        messageMap.clear();
        queueMessageMap.clear();
        // 2. 恢复所有的交换机数据
        List<Exchange> exchanges = diskDataCenter.selectAllExchanges();
        for (Exchange exchange : exchanges){
            exchangeMap.put(exchange.getName(),exchange);
        }
        // 3. 恢复所有的队列数据
        List<MSGQueue> queues = diskDataCenter.selectAllMSGQueues();
        for (MSGQueue queue : queues){
            queueMap.put(queue.getName(),queue);
        }
        // 4. 恢复所有的绑定数据
        List<Binding> bindings = diskDataCenter.selectAllBindings();
        for (Binding binding : bindings){
            ConcurrentHashMap<String,Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(),
                    k -> new ConcurrentHashMap<>());
            bindingMap.put(binding.getQueueName(),binding);
        }
        // 5. 恢复所有的消息数据
        // 遍历所有的队列, 根据所有的队列名, 获得所有的信息
        for (MSGQueue queue : queues){
            LinkedList<Message> messages = diskDataCenter.loadAllMessageFromQueue(queue.getName());
            queueMessageMap.put(queue.getName(),messages);
            for (Message message : messages){
                messageMap.put(message.getMessageId(),message);
            }
        }

        // 注意!! 针对 "未被确认的消息" 这部分内存的数据, 不需要从硬盘中去恢复, 之前考虑硬盘存储的时候, 未涉及这一块
        // 一旦在等待 ack 的过程中, 服务器被重启了, 此时这些 "未被确认的消息", 就会自动恢复成 "未被取走的消息"
        // 这个消息在硬盘中都是 "未取走的消息", 就当当成是未取走的
    }
}
